package cn.itcast.flink.connector;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * @author lilulu
 */
public class ConnectorFlinkKafkaConsumerOffsetDemo {
    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 数据源-source
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "node1:9092,node2:9092,node3:9092,");
        properties.setProperty("group.id", "test1");
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<String>(
                //新增一类数据，同时新增1个 Kafka topic，如何在==不重启作业==的情况下作业自动感知新的 topic。
                java.util.regex.Pattern.compile("test-topic-[0-9]"),
//                "flink-topic",
                new SimpleStringSchema(),
                properties
        );

        // TODO: 1、Flink从topic中最初的数据开始消费
//        kafkaConsumer.setStartFromEarliest() ;
        // TODO: 2、Flink从topic中最新的数据开始消费
        //kafkaConsumer.setStartFromLatest();
        // TODO: 3、Flink从topic中指定的group上次消费的位置开始消费，所以必须配置group.id参数
        //kafkaConsumer.setStartFromGroupOffsets() ;
        // TODO: 4、Flink从topic中指定的offset开始，这个比较复杂，需要手动指定offset
       /* Map<KafkaTopicPartition, Long> offsets = new HashMap<>();
        offsets.put(new KafkaTopicPartition("flink-topic", 0), 100L);
        offsets.put(new KafkaTopicPartition("flink-topic", 1), 90L);
        offsets.put(new KafkaTopicPartition("flink-topic", 2), 110L);*/
        //kafkaConsumer.setStartFromSpecificOffsets(offsets);
        // TODO: 5、指定时间戳消费数据
//        kafkaConsumer.setStartFromTimestamp(1644935966961L);

        DataStreamSource<String> kafkaDataStream = env.addSource(kafkaConsumer);
        // 3. 数据转换-transformation
        // 4. 数据终端-sink
        kafkaDataStream.print();
        // 5. 触发执行-execute
        env.execute("ConnectorFlinkKafkaConsumerOffsetDemo");
    }
}